package com.xinyue.game.gateway.rocketmq;

import java.io.IOException;
import java.util.List;

import javax.annotation.PostConstruct;
import javax.annotation.PreDestroy;

import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.MessageQueueSelector;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.common.message.MessageQueue;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

import com.xinyue.game.gateway.common.GameGatewayConfig;

import io.netty.buffer.ByteBuf;

/**
 * 消息队列的客户端服务，用来向消息队列服务发送消息
 *
 * @author 王广帅
 * @since 2023/6/28 20:55
 **/
@Service
public class RocketMQClientService {

    private static final Logger logger = LoggerFactory.getLogger(RocketMQClientService.class);
    @Autowired
    private GameGatewayConfig gameGatewayConfig;

    private DefaultMQProducer mqProducer;

    @PostConstruct
    public void init() {
        String endpoint = gameGatewayConfig.getRocketmqServer();

        // 初始化Producer时需要设置通信配置以及预绑定的Topic。
        try {
            String groupName = gameGatewayConfig.getGatewayGroupName();
            mqProducer = new DefaultMQProducer(groupName);
            mqProducer.setNamesrvAddr(gameGatewayConfig.getRocketmqServer());
            mqProducer.start();
            logger.info("MQ生产者启动成功!!,host: {}", endpoint);
        } catch (Throwable e) {
            logger.error("初始化RocketMQ客户端失败,endPoint: {}", endpoint, e);
            throw new IllegalArgumentException("初始化RocketMQ客户端失败");
        }
    }

    public String getTopic(String gameId, String serviceId) {
        return gameGatewayConfig.getLogicServerTopicPrefix() + "_" + gameId + "_" + serviceId;
    }


    /**
     * 向消息队列中发送消息
     *
     * @param key     关键字，用来选择队列，相同的key都会存到同一个队列之中
     * @param topic   消息队列主题名称
     * @param byteBuf 要发送的消息
     */
    public void sendMsg(String key, String topic, String tag, ByteBuf byteBuf) {
        int len = byteBuf.readableBytes();
        byte[] bytes = new byte[len];
        byteBuf.readBytes(bytes);
        this.sendMsg(key, topic, tag, bytes);
    }

    /**
     * 向消息队列中发送消息
     *
     * @param key     关键字，用来选择队列，相同的key都会存到同一个队列之中
     * @param topic   消息队列主题名称
     * @param byteBuf 要发送的消息
     */
    public void sendMsg(String key, String topic, String tag, byte[] byteBuf) {
        // 普通消息发送。
        Message message = new Message(topic, tag, byteBuf);
        try {
            mqProducer.send(message, new MessageQueueSelector() {
                @Override
                public MessageQueue select(List<MessageQueue> list, Message message, Object o) {
                    int size = list.size();
                    int hash = Math.abs(key.hashCode());
                    int index = hash % size;
                    return list.get(index);
                }
            }, 1000);
        } catch (Throwable e) {
            logger.error("mq发送消息异常", e);
        }
    }

    @PreDestroy
    private void stopServer() throws IOException {
        if (mqProducer != null) {
            mqProducer.shutdown();
        }
    }
}
